package com.jbm.core.common;

import com.google.protobuf.ByteString;
import com.jbm.core.common.ResponseRecvCallback.ResponseRecvCallback;
import com.jbm.core.iface.IAcross;
import com.jbm.core.pb.PBIAcross;
import com.jbm.core.pb.PBIP2P;
import com.mongodb.MongoClient;
import com.mongodb.client.FindIterable;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.MongoDatabase;
import com.mongodb.client.model.Filters;
import com.mongodb.client.result.UpdateResult;

import org.bson.Document;
import org.bson.conversions.Bson;
import org.bson.types.Binary;
import org.bson.types.ObjectId;

/**
 * Created by root on 17-11-11.
 */
class MQThread extends Thread
{
	@Override
	public void run()
	{
		while (true)
		{
			try
			{
				PBIAcross.IAcross req = MSGQueue3MongoDB.take(null);

				if ( req != null )
					AutoCall.g_autocall.Call(req, MSGQueue3MongoDB.m_rrcallback, false, 0, null, MainServer.Server_type_Server);

				sleep(5);
			}catch (Exception e)
			{
			}
		}
	}
}

public class MSGQueue3MongoDB
{
	private static MongoClient mongoClient = null;
	private static MongoDatabase mongoDatabase = null;
	public static ResponseRecvCallback m_rrcallback = null;
	private static MongoCollection<Document> send_collection = null;
	private static MongoCollection<Document> dlog_collection = null;
	private static MongoCollection<Document> opas_collection = null;
	private static MongoCollection<Document> offl_collection = null;
	private static MongoCollection<Document> pic_collection = null;

	static
	{
		try
		{
			mongoClient = new MongoClient("localhost", 27017);

			mongoDatabase = mongoClient.getDatabase("jsm_p2p");

			send_collection = mongoDatabase.getCollection("msgq");
			dlog_collection = mongoDatabase.getCollection("dlog");
			opas_collection = mongoDatabase.getCollection("opas");
			offl_collection = mongoDatabase.getCollection("offl");
			pic_collection = mongoDatabase.getCollection("pices");

			m_rrcallback = new ResponseRecvCallback()
			{
				public void PushResponse(long start, long cur, PBIAcross.IAcross request, IAcross content, PBIAcross.IAcross response, Object exinfo, boolean isfirst)
				{
					if ( response == null )
						return;

					try
					{
						if (response.getCode() != 200)
							put(exinfo, false);
						else
							save((PBIAcross.IAcross)exinfo);
					}catch (Exception e)
					{
					}
				}
			};

			for ( int i = 0; i < 20; i ++ )
			{
				MQThread t = new MQThread();
				t.start();
			}
		}catch (Exception e)
		{
		}
	}

	public static void Start()
	{
	}

	public static PBIAcross.IAcross take(PBIAcross.IAcross req)
	{
		if ( req == null )
		{
			Document dreq = send_collection.findOneAndDelete(new Document());

			if (dreq != null)
			{
//				PBIAcross.IAcross req = null;
				try
				{
					Binary bbody = (Binary) dreq.get("body");

					dreq.remove("_id");
					dreq.remove("body");

					opas_collection.insertOne(dreq);

					req = PBIAcross.IAcross.parseFrom(bbody.getData());

					byte[] body = (req = req.toBuilder().setMongodbid(ByteString.copyFrom(dreq.getObjectId("_id").toByteArray())).build()).toByteArray();
					opas_collection.updateOne(Filters.eq("_id", dreq.getObjectId("_id")), new Document("$set", new Document("body", body)));
				} catch (Exception e)
				{
				}

				return req;
			} else
				return null;
		}else
		{
			Document dreq = new Document("fromid", req.getFromid()).
					append("toid", req.getToid()).
					append("fromtag", req.getFromtag()).
					append("totag", req.getTotag()).
					append("class", req.getClass_()).
					append("method", req.getMethod()).
					append("readed",false).
					append("timestamp",req.getTimestamp());

			try
			{
				if (req.getClass_().equals("IP2P"))
				{
					PBIP2P.IP2P p2p = PBIP2P.IP2P.parseFrom(req.getContent());
					dreq.append("msg", p2p.getMMsg());
					dreq.append("order_id",p2p.getOrderId());
					dreq.append("task_id",p2p.getTaskId());
					dreq.append("sc_type",p2p.getType());
				}
			}catch (Exception e)
			{
			}

			opas_collection.insertOne(dreq);

			byte[] body = (req = req.toBuilder().setMongodbid(ByteString.copyFrom(dreq.getObjectId("_id").toByteArray())).build()).toByteArray();
			opas_collection.updateOne(Filters.eq("_id", dreq.getObjectId("_id")), new Document("$set", new Document("body", body)));

			return req;
		}
	}

	public static void put(Object obj, boolean isfirst)
	{
		PBIAcross.IAcross req = (PBIAcross.IAcross)obj;

		Document dreq = null;

		if ( !isfirst )
		{
			dreq = opas_collection.findOneAndDelete(
					Filters.eq("_id", new ObjectId(req.getMongodbid().toByteArray())));

			dreq.remove("_id");
		}else
		{
			dreq = new Document("fromid", req.getFromid()).
					append("toid", req.getToid()).
					append("fromtag", req.getFromtag()).
					append("totag", req.getTotag()).
					append("body",req.toByteArray());
		}

		send_collection.insertOne(dreq);
	}

	private static void save(PBIAcross.IAcross req)
	{
		Document dreq = opas_collection.findOneAndDelete(
				Filters.eq("_id", new ObjectId(req.getMongodbid().toByteArray())));

		if ( dreq != null )
		{
			dreq.remove("_id");
			dlog_collection.insertOne(dreq);
		}
	}

	public static FindIterable<Document> getDLogList(Bson ft,int sort)
	{
/*
		Bson ft = null;
		if ( fromid > 0 && toid > 0 )
			ft = Filters.or(Filters.eq("fromid",fromid),Filters.eq("toid",toid));
		else if ( fromid > 0 )
			ft = Filters.eq("fromid",fromid);
		else
			ft = Filters.eq("toid",toid);
*/

		return dlog_collection.find(ft).sort( new Document("timestamp",sort));
	}

	public static boolean MarkMessageReaden(byte[] id)
	{
		UpdateResult ur = null;

		try
		{
			ur = dlog_collection.updateOne(Filters.eq("_id", new ObjectId(id)), new Document("$set", new Document("readed", true)));
		}catch (Exception e)
		{
//			e.printStackTrace();
		}

		return ur.getMatchedCount() > 0;
	}

	public static void SaveIDCard(byte[] face,byte[] back,int user_id)
	{
		pic_collection.insertOne(new Document("user_id",user_id).
				append("id_card_face",face).
				append("id_card_back",back).
				append("timestamp",System.currentTimeMillis()));
	}

//	public static Document getIDCard(int user_id)
}